package org.smartboot.sosa.rpc.netty;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.smartboot.sosa.core.util.StringUtils;
import org.smartboot.sosa.rpc.RpcConstant;
import org.smartboot.sosa.rpc.serial.SerializableBean;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.EncoderException;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.util.AttributeKey;
import io.protostuff.LinkedBuffer;
import io.protostuff.ProtobufIOUtil;
import io.protostuff.Schema;
import io.protostuff.runtime.RuntimeSchema;

/**
 * RPC消息编码
 * 
 * @author Seer
 * @version RpcEncoder.java, v 0.1 2015年11月21日 上午10:57:54 Seer Exp.
 */
public class RpcEncoder extends MessageToByteEncoder<SerializableBean> {
	private static final String LINK_BUFFER = "buffer";
	private static final Logger LOGGER = LogManager.getLogger(RpcEncoder.class);

	@Override
	protected void encode(ChannelHandlerContext ctx, SerializableBean bean, ByteBuf out) throws Exception {
		if (bean.getBean() == null) {
			throw new EncoderException("RPC message is null,can not encode!");
		}
		Schema<SerializableBean> schema = RuntimeSchema.getSchema(SerializableBean.class);
		Channel channel = ctx.channel();

		AttributeKey<LinkedBuffer> key = AttributeKey.valueOf(LINK_BUFFER);
		LinkedBuffer buffer = channel.attr(key).get();
		if (buffer == null) {
			buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
			channel.attr(key).set(buffer);
		}

		byte[] bytes = ProtobufIOUtil.toByteArray(bean, schema, buffer);
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("encode bit stream " + StringUtils.toHexString(bytes));
		}
		out.writeBytes(RpcConstant.MAGIC_NUM).writeInt(bytes.length).writeBytes(bytes);

		buffer.clear();
	}

}
